package com.atguigu.bigdata.spark.core.rdd.req;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.sources.In;
import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple4;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

public class Spark01_Req1_HotCategoryTop10Analysis_JAVA {
    public static void main(String[] args) {
        /*
            本项目需求优化为：先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数
         */
        // 1.创建配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis");
        // 2. 创建sparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        //读取原始日志数据

        JavaRDD<String> rdd = sc.textFile("datas/user_visit_action.txt");

        // 统计品类的点击数量：（品类ID，点击数量）
        //  排除非点击数据
        JavaRDD<String>  filterRDD = rdd.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[6].equals("-1"));
            }
        });
        //转换格式 (品类ID，点击数量）
        JavaPairRDD<String, Integer> pairRDD = filterRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                String[] temp = s.split("_");
                return new Tuple2<String, Integer>(temp[6], 1);
            }
        });

        JavaPairRDD<String, Integer> reducebykeyClick = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        //-------------------------------------------
        // 统计品类的下单数量：（品类ID，下单数量）

        JavaRDD<String> orderRDD = rdd.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[8].contains("null"));
            }
        });
        //        // orderid => 1,2,3
        //        // 【(1,1)，(2,1)，(3,1)】
        JavaPairRDD<String, Integer> reducebykeyOrder = orderRDD.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] temp = s.split("_");
                List<Tuple2<String, Integer>> list = new ArrayList<>();
                String[] temp1 = temp[8].split(",");
                for(String val : temp1) {
                    list.add(new Tuple2<String, Integer>(val ,1));
                }
                return list.iterator();
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        //----------------------------------------------------------------------------
        //统计品类的支付数量：（品类ID，支付数量）
        JavaRDD<String>  payActionRDD = rdd.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                String[] temp = s.split("_");
                return (!temp[10].contains("null"));
            }
        });
        JavaPairRDD<String, Integer> reducebykeyPayment = orderRDD.flatMapToPair(new PairFlatMapFunction<String, String, Integer>() {
            @Override
            public Iterator<Tuple2<String, Integer>> call(String s) throws Exception {
                String[] temp = s.split("_");
                List<Tuple2<String, Integer>> list = new ArrayList<>();
                String[] temp1 = temp[10].split(",");
                for(String val : temp1) {
                    list.add(new Tuple2<String, Integer>(val ,1));
                }
                return list.iterator();
            }
        }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        // 5. 将品类进行排序，并且取前10名
        //    点击数量排序，下单数量排序，支付数量排序
        //    元组排序：先比较第一个，再比较第二个，再比较第三个，依此类推
        //    ( 品类ID, ( 点击数量, 下单数量, 支付数量 ) )
        //
        JavaPairRDD<String, Tuple3<Integer,Integer,Integer>> analysis =  reducebykeyClick.cogroup(reducebykeyOrder,reducebykeyPayment).mapValues(new Function<Tuple3<Iterable<Integer>, Iterable<Integer>, Iterable<Integer>>, Tuple3<Integer, Integer, Integer>>() {
            @Override
            public Tuple3<Integer, Integer, Integer> call(Tuple3<Iterable<Integer>, Iterable<Integer>, Iterable<Integer>> v1) throws Exception {
                int clickCnt = 0;
                int orderCnt = 0;
                int payCnt = 0;
                Iterator<Integer> click = v1._1().iterator();
                Iterator<Integer> order = v1._2().iterator();
                Iterator<Integer> pay = v1._3().iterator();
                if (click.hasNext()) {
                    clickCnt = click.next();
                }
                if (order.hasNext()) {
                    orderCnt = order.next();
                }
                if (pay.hasNext()) {
                    payCnt = pay.next();
                }
                return new Tuple3<Integer, Integer, Integer>(clickCnt,orderCnt,payCnt);
            }
        });
        JavaRDD<Tuple4<String, Integer, Integer, Integer>> analysis_map = analysis.map(new Function<Tuple2<String, Tuple3<Integer, Integer, Integer>>, Tuple4<String, Integer, Integer, Integer>>() {
            @Override
            public Tuple4<String, Integer, Integer, Integer> call(Tuple2<String, Tuple3<Integer, Integer, Integer>> v1) throws Exception {
                Tuple3<Integer, Integer, Integer> t3 = v1._2;
                return new Tuple4<String, Integer, Integer, Integer>(v1._1, t3._1(), t3._2() ,t3._3());
            }
        });

        analysis_map.sortBy(new Function<Tuple4<String, Integer, Integer, Integer>, Object>() {
            @Override
            public Object call(Tuple4<String, Integer, Integer, Integer> v1) throws Exception {
                return v1._2();
            }
        },false,5);

        System.out.println(analysis_map.take(10).toString());

        sc.stop();

    }
}
